23. 物流行业数据分析

import pandas as pd  # 导入表格库处理平台订单表
import numpy as np  # 导入有限值与整数检查要求工具
def validate_platform_quantities(logistics_orders):  # 定义销售数量质量要求
    required={'销售时间','销售区域','货品','货品交货状况','货品用户反馈','数量'}  # 对齐平台中文字段要求
    if not required<=set(logistics_orders.columns): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 缺列时停止
    quantity_numeric=pd.to_numeric(logistics_orders['数量'],errors='coerce')  # 将数量显式解析为数值
    quantity_audit={'missing':int(logistics_orders['数量'].isna().sum())}  # 统计原始缺失数量
    quantity_audit['nonnumeric']=int(logistics_orders['数量'].notna().mul(quantity_numeric.isna()).sum())  # 统计非数字数量
    finite_mask=quantity_numeric.notna()&np.isfinite(quantity_numeric.fillna(0))  # 标记可继续检查的有限值
    quantity_audit['nonfinite']=int(quantity_numeric.notna().mul(~np.isfinite(quantity_numeric.fillna(0))).sum())  # 统计无穷值
    quantity_audit['nonintegral']=int(finite_mask.mul(~np.isclose(quantity_numeric.fillna(0),np.round(quantity_numeric.fillna(0)))).sum())  # 统计非整数
    quantity_audit['nonpositive']=int(finite_mask.mul(quantity_numeric.fillna(0).le(0)).sum())  # 统计非正数量
    quantity_audit['policy']='销售数量必须为正整数；退货须用单独字段记录'  # 记录数量处理规则
    if any(quantity_audit[key] for key in ['missing','nonnumeric','nonfinite','nonintegral','nonpositive']): raise ValueError(f'输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 任一违规即停止
    prepared_orders=logistics_orders.assign(数量=quantity_numeric.astype('int64'))  # 检查要求通过后固定整数类型
    return prepared_orders,quantity_audit  # 返回合格订单与质量检查

def summarize_platform_logistics(logistics_orders):  # 汇总同一平台订单快照
    prepared_orders,quantity_audit=validate_platform_quantities(logistics_orders)  # 先执行数量质量要求
    required=['销售时间','销售区域','货品','货品交货状况','货品用户反馈','数量']  # 固定聚合字段
    clean_orders=prepared_orders.dropna(subset=required).drop_duplicates().copy()  # 排除关键缺失与完全重复
    if clean_orders.empty: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 空样本时停止
    clean_orders['销售时间']=pd.to_datetime(clean_orders['销售时间'])  # 统一销售时间类型
    clean_orders['销售月份']=clean_orders['销售时间'].dt.to_period('M').astype(str)  # 构造月度键
    monthly=clean_orders.groupby('销售月份').agg(rows=('数量','size'),quantity=('数量','sum'))  # 汇总月度行数与销量
    regional=clean_orders.groupby(['销售区域','货品交货状况']).size().unstack(fill_value=0)  # 汇总地区配送状态
    product=clean_orders.groupby(['货品','货品交货状况']).size().unstack(fill_value=0)  # 汇总货品配送状态
    feedback=clean_orders.groupby(['货品','销售区域','货品用户反馈']).size().unstack(fill_value=0)  # 统计分类反馈
    feedback_rate=feedback.div(feedback.sum(axis=1),axis=0)  # 按货品与地区分母计算反馈占比
    if feedback_rate.isna().any().any(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 零分母时停止
    first_region=sorted(clean_orders['销售区域'].astype(str).unique())[0]  # 固定首个地区核对
    region_slice=clean_orders.loc[clean_orders['销售区域'].astype(str).eq(first_region)].groupby('货品')['数量'].sum()  # 独立复算地区货品量
    clean_quantity=clean_orders['数量'].sum()  # 独立计算清洗后总销量
    if clean_quantity!=monthly['quantity'].sum(): raise AssertionError('quantity reconciliation failed')  # 核对月度汇总守恒
    region_quantity=clean_orders.loc[clean_orders['销售区域'].astype(str).eq(first_region),'数量'].sum()  # 独立计算地区总量
    if region_quantity!=region_slice.sum(): raise AssertionError('region quantity reconciliation failed')  # 核对地区分组守恒
    reconciliation={'raw_rows':len(logistics_orders),'clean_rows':len(clean_orders),'clean_quantity':clean_quantity,'region':first_region,'region_quantity':region_quantity}  # 形成检查摘要
    return clean_orders,monthly,regional,product,feedback_rate,region_slice,reconciliation,quantity_audit  # 返回聚合与检查依据

from pathlib import Path  # 导入路径工具绑定规定根目录
import pandas as pd  # 导入表格库以读取并标准化HDF字段
import numpy as np  # 导入有限值检查要求工具
root=Path('/home/ubuntu/r2_data_mount/data/stock')  # 指定中国股票数据目录
basic_path=root/'stock_basic_data.h5'  # 绑定公司基础信息HDF快照
price_path=root/'stock_price_pre_adjusted.h5'  # 绑定前复权行情HDF快照
if not basic_path.exists() or not price_path.exists(): raise FileNotFoundError('未找到课程数据文件，请从课程数据下载入口获取并核对文件位置')  # 缺失时如实停止
basic=pd.read_hdf(basic_path,key='stock_basic_info',columns=['order_book_id','symbol','industry_name','province']).rename(columns={'order_book_id':'raw_code','symbol':'name','industry_name':'industry','province':'area'})  # 只读取对象筛选字段并标准化
basic['ts_code']=basic['raw_code'].str.replace('.XSHG','.SH',regex=False).str.replace('.XSHE','.SZ',regex=False)  # 统一证券代码后缀
basic['area']=basic['area'].astype('string').str.replace(r'[省市]$','',regex=True)  # 统一省市简称
required_basic={'ts_code','name','industry','area'}  # 定义基础信息字段要求
if not required_basic<=set(basic.columns): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 字段结构 不符时停止
industry_pattern='交通运输|仓储|邮政|快递|道路运输|水上运输|航空运输|装卸搬运|多式联运'  # 排除运输设备制造等非物流运营行业
focus=basic.loc[basic['industry'].str.contains(industry_pattern,na=False) & basic['area'].isin(['上海','江苏','浙江','安徽'])].sort_values('ts_code').head(12)  # 按固定规则选取长三角物流运营公司
if focus.empty: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 空对象时停止
focus_raw_codes=focus['raw_code'].tolist()  # 保留HDF原始代码以高效筛选行情
prices=pd.read_hdf(price_path,key='data',where='order_book_id in focus_raw_codes',columns=['close','volume']).reset_index().rename(columns={'order_book_id':'ts_code','date':'trade_date','volume':'vol'})  # 按对象读取并标准化行情字段
prices['ts_code']=prices['ts_code'].str.replace('.XSHG','.SH',regex=False).str.replace('.XSHE','.SZ',regex=False)  # 统一行情证券代码后缀
prices=prices.loc[pd.to_datetime(prices['trade_date']).between('2019-01-01','2025-12-31')].copy()  # 固定观察期，便于比较结果
market=prices.merge(focus[['ts_code','name','area']],on='ts_code',how='inner',validate='many_to_one')  # 合并行情与地区标签

if focus.empty or market.empty: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 空对象或空连接时停止
if not np.isfinite(market[['close','vol']].to_numpy()).all() or market['close'].le(0).any(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 行情必须有限且价格为正
if market.duplicated(['ts_code','trade_date']).any(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 重复键时停止
market['trade_date']=pd.to_datetime(market['trade_date'])  # 统一交易日期
market['month']=market['trade_date'].dt.to_period('M').astype(str)  # 构造月度观察窗
if market['month'].nunique()<6: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 期间不足时停止趋势判断
market['return_rate']=market.groupby('ts_code')['close'].pct_change()  # 计算公司日收益率
monthly=market.groupby(['month','area']).agg(companies=('ts_code','nunique'),mean_return=('return_rate','mean'),trading_volume=('vol','sum')).reset_index()  # 输出月度地区聚合
sensitivity=monthly.assign(loose=monthly['mean_return'].abs()>0.02,baseline=monthly['mean_return'].abs()>0.03,strict=monthly['mean_return'].abs()>0.05,threshold_status='课堂情景，待业务规定批准',owner='物流分析负责人',action='核对异常月份',recheck_when='字段发生变化或分母为零',recheck_date='下月第5个工作日')  # 比较三档课堂阈值并绑定说明信息
assert monthly['companies'].gt(0).all()  # 核对每个聚合单元具有明确分母
print(focus[['ts_code','name','area']],monthly,sensitivity)  # 输出对象、聚合与检查要求表